package com.lyrcsoft.rokectmq.details;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * @author fujc2dev@126.com
 * @date 2024-05-11 11:03
 */
public class ProducerDetails {
    public static void main(String[] args) throws Exception {
        // todo  producerGroup：生产者所属组（主要针对事务消息，高可用）
        DefaultMQProducer producer = new DefaultMQProducer("xxx");
        //  默认主题在每一个broker队列数量（对于创建新主题有效）
        // 主题在Broker里面的队列数量
        producer.setDefaultTopicQueueNums(8);
        // 发送消息默认超时时间。默认3s  (3000ms)
        producer.setSendMsgTimeout(1000 * 3);
        // 消息体超过该值则启用压缩，默认4K
        producer.setCompressMsgBodyOverHowmuch(1024 * 4);
        // 同步方式发送消息重试次数，默认2，总共执行3次
        producer.setRetryTimesWhenSendFailed(2);
        //producer.setRetryTimesWhenSendAsyncFailed(2);  异步
        // 消息重试时，选择另外一个broker发送，（消息没有存储成功是否发送到另外一个broker），默认false
        // 什么意思？就是在双主节点切换时，如果发送到第一个节点失败，则发送到第二个节点，设置其不允许。
        producer.setRetryAnotherBrokerWhenNotStoreOK(false);
        //  允许发送的最大消息长度，默认4M，阿里人员经过大量分需后，得出的最大值是4M
        producer.setMaxMessageSize(1024 * 1024 * 4);
        producer.setNamesrvAddr("10.20.200.144:9876");
        producer.start();
        // 查找该主题下的所有消息队列
        List<MessageQueue> messageQueue = producer.fetchPublishMessageQueues("TopicTest");
        for (int i = 0; i < messageQueue.size(); i++) {
            System.out.println(messageQueue.get(i).getQueueId());
        }

        Message message = new Message("TopicTest", "", "", "hello word".getBytes(RemotingHelper.DEFAULT_CHARSET));
        // 单向发送，发送了就不管了
        // 1.1 发送单向消息
        producer.sendOneway(message);
        // 1.2 指定队列单向发送消息，使用select选择器队列
        producer.sendOneway(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                return mqs.get(0);
            }
        }, null);
        // 1.3 指定队列发送消息，根据之前查询出来的队列
        producer.sendOneway(message, messageQueue.get(0));

        // 同步发送
        // 2,1 同步发现消息
        SendResult sendResult0 = producer.send(message);
        // 2.2 同步超时发送消息，sendtime 发送消息默认超时时间 默认3s
        SendResult sendResult1 = producer.send(message, 1000 * 3);
        // 2.3 指定队列同步发送消息，使用select选择器队列
        SendResult sendResult2 = producer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                return mqs.get(0);
            }
        }, null);
        // 2.4 指定队列同步发送消息，根据之前查询出来的队列
        SendResult sendResult3 = producer.send(message, messageQueue.get(0));
        Integer index = null;
        // 异步发送
        // 3.1 异步发送消息
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable e) {
                System.out.printf("%-10d OK %s %n", index, e);
                e.printStackTrace();
            }
        });
        // 3.2 异步超时发送消息
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable e) {
                System.out.printf("%-10d OK %s %n", index, e);
                e.printStackTrace();
            }
        }, 1000 * 3);
        // 3.3 选择指定队列异步发送消息
        producer.send(message, messageQueue.get(0), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable e) {
                System.out.printf("%-10d OK %s %n", index, e);
            }
        });
        // 3.4 指定队列选择器异步发送消息
        producer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                return mqs.get(0);
            }
        }, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable e) {
                System.out.printf("%-10d OK %s %n", index, e);
            }
        });
        producer.shutdown();
    }
}
